package com.midgetontoes.reactorexamples.youtube;

import reactor.core.Disposable;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: zhangQi
 * @Date: 2020-10-11 11:59
 */
public class FlightOfTheFlux {
    /**
     * Array of names to use in demo
     */
    private static String[] NAMES = new String[]{
            "Julie",
            "Joe",
            "Simon",
            "Bob"
    };

    private static String indexToName(Integer i) {
        System.out.println("finding a name for " + i + " from thread " + Thread.currentThread().getName());
        return NAMES[i];
    }

    public static void main(String[] args) throws Exception {
        //TODO 1. Nothing happens until you subscribe
        Flux<String> flux = Flux.range(0, 4)
                .map(FlightOfTheFlux::indexToName);
//        flux.subscribe(v -> System.out.println(v));

        System.out.println();
        //TODO 2. Hot vs Cold
        flux.subscribe(v -> System.out.println("first:" + v), Throwable::printStackTrace);
        flux.subscribe(v -> System.out.println("second:" + v), Throwable::printStackTrace);

        System.out.println("---------happyLine---------");

        AtomicInteger count = new AtomicInteger();

        final Flux<Object> generate = Flux.generate(sink -> sink.next(count.incrementAndGet()));

        generate.take(4).subscribe(v -> System.out.println("first received " + v), Throwable::printStackTrace);
        generate.take(4).subscribe(v -> System.out.println("second received " + v), Throwable::printStackTrace);

        System.out.println("---------happyLine---------");

//        final ConnectableFlux<String> cflux = flux.publish();
        final ConnectableFlux<String> cflux = flux.delayElements(Duration.ofMillis(500)).publish();
        cflux.subscribe(v -> System.out.println("first cflux : " + v), Throwable::printStackTrace);
        cflux.connect();
        Thread.sleep(1500);
        cflux.subscribe(v -> System.out.println("second cflux : " + v), Throwable::printStackTrace);
        System.out.println("---------happyLine---------");
        //TODO 3. PublishOn
        Scheduler s = Schedulers.elastic();
//        flux.publishOn(s)
//                .subscribe(v-> System.out.println("finally received "+v+" on thread "+Thread.currentThread().getName()));

        flux.publishOn(s)
                .doOnNext(v -> System.out.println("finally received " + v + " on thread " + Thread.currentThread().getName()))
                .subscribe();
//        flux.subscribe(System.out::println);
        System.out.println("---------happyLine---------");
        //TODO 4. SubscribeOn
//        flux.subscribeOn(s)
//                .doOnNext(v -> System.out.println("finally subscribeOn received " + v + " on thread " +
//                        Thread.currentThread().getName()))
//                .subscribe();
        Flux.range(0, 4)
                .map(FlightOfTheFlux::indexToName)
                .subscribeOn(s)
                .doOnNext(v -> System.out.println("finally subscribeOn received " + v + " on thread " + Thread.currentThread().getName()))
                .subscribe();

        System.out.println();

    }

}
